/*
 * Unidata Platform Community Edition
 * Copyright (c) 2013-2020, UNIDATA LLC, All rights reserved.
 * This file is part of the Unidata Platform Community Edition software.
 *
 * Unidata Platform Community Edition is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Unidata Platform Community Edition is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <https://www.gnu.org/licenses/>.
 */

package org.unidata.mdm.dq.data.service.segments.records.get;

import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.unidata.mdm.core.service.MetaModelService;
import org.unidata.mdm.core.type.model.EntityElement;
import org.unidata.mdm.data.configuration.DataNamespace;
import org.unidata.mdm.data.context.GetRequestContext;
import org.unidata.mdm.data.type.keys.RecordKeys;
import org.unidata.mdm.dq.core.configuration.DataQualityDescriptors;
import org.unidata.mdm.dq.core.dto.DataQualityResult;
import org.unidata.mdm.dq.core.dto.DataQualityResult.RuleExecutionResult;
import org.unidata.mdm.dq.core.serialization.DataQualitySerializer;
import org.unidata.mdm.dq.core.type.io.DataQualityError;
import org.unidata.mdm.dq.core.type.io.DataQualitySpot;
import org.unidata.mdm.dq.core.type.model.instance.MappingSetElement;
import org.unidata.mdm.dq.core.type.model.instance.NamespaceAssignmentElement;
import org.unidata.mdm.dq.core.type.model.instance.QualityRuleElement;
import org.unidata.mdm.dq.data.dto.RecordGetQualityResult;
import org.unidata.mdm.dq.data.module.DataQualityDataModule;
import org.unidata.mdm.dq.data.type.search.RecordsDataQualityHeaderField;
import org.unidata.mdm.meta.configuration.Descriptors;
import org.unidata.mdm.meta.type.search.EntityIndexType;
import org.unidata.mdm.meta.type.search.RecordHeaderField;
import org.unidata.mdm.search.context.NestedSearchRequestContext;
import org.unidata.mdm.search.context.SearchRequestContext;
import org.unidata.mdm.search.dto.SearchResultDTO;
import org.unidata.mdm.search.dto.SearchResultHitDTO;
import org.unidata.mdm.search.service.SearchService;
import org.unidata.mdm.search.type.form.FieldsGroup;
import org.unidata.mdm.search.type.form.FormField;
import org.unidata.mdm.search.type.query.SearchQuery;
import org.unidata.mdm.system.type.namespace.NameSpace;
import org.unidata.mdm.system.type.pipeline.Connector;
import org.unidata.mdm.system.type.pipeline.Pipeline;
import org.unidata.mdm.system.type.pipeline.Start;
import org.unidata.mdm.system.type.runtime.MeasurementPoint;

/**
 * @author Mikhail Mikhailov on Nov 24, 2019
 */
@Component(RecordGetQualityConnectorExecutor.SEGMENT_ID)
public class RecordGetQualityConnectorExecutor extends Connector<GetRequestContext, RecordGetQualityResult> {
    /**
     * This segment ID.
     */
    public static final String SEGMENT_ID = DataQualityDataModule.MODULE_ID + "[RECORD_GET_QUALITY_CONNECTOR]";
    /**
     * Localized message code.
     */
    public static final String SEGMENT_DESCRIPTION = DataQualityDataModule.MODULE_ID + ".record.get.quality.connector.description";
    /**
     * The MMS instance.
     */
    @Autowired
    private MetaModelService metaModelService;
    /**
     * The SS.
     */
    @Autowired
    private SearchService searchService;
    /**
     * Constructor.
     * @param id
     * @param description
     */
    public RecordGetQualityConnectorExecutor() {
        super(SEGMENT_ID, SEGMENT_DESCRIPTION);
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public RecordGetQualityResult connect(GetRequestContext ctx) {
        return execute(ctx, null);
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public RecordGetQualityResult connect(GetRequestContext ctx, Pipeline p) {
        return execute(ctx, p);
    }
    /**
     * Gets DQ Data data.
     * @param ctx the context
     * @param p the pipeine (we don't care)
     * @return result
     */
    public RecordGetQualityResult execute(@Nonnull GetRequestContext ctx, @Nullable Pipeline p) {

        MeasurementPoint.start();
        try {

            RecordGetQualityResult result = new RecordGetQualityResult();
            RecordKeys keys = ctx.keys();
            if (keys != null) {

                EntityElement el = metaModelService.instance(Descriptors.DATA)
                    .getElement(keys.getEntityName());

                NameSpace target = null;
                if (el.isLookup()) {
                    target = DataNamespace.LOOKUP;
                } else if (el.isRegister()) {
                    target = DataNamespace.REGISTER;
                }

                NamespaceAssignmentElement nae = metaModelService.instance(DataQualityDescriptors.DQ)
                    .getAssignment(target);

                if (nae == null || !nae.isAssigned(keys.getEntityName())) {
                    return result;
                }

                SearchResultDTO search = executeSearch(keys, ctx);
                if (CollectionUtils.isEmpty(search.getHits())
                 || CollectionUtils.isEmpty(search.getHits().get(0).getInnerHits().get(RecordsDataQualityHeaderField.FIELD_QUALITY_ERRORS.getPath()))) {
                    return result;
                }

                result.setPayload(extractSearch(search));
            }

            return result;
        } finally {
            MeasurementPoint.stop();
        }
    }

    private SearchResultDTO executeSearch(RecordKeys keys, GetRequestContext gCtx) {

        Date point = gCtx.getForDate() == null ? new Date() : gCtx.getForDate();
        FieldsGroup and = FieldsGroup.and()
                .add(FormField.exact(RecordHeaderField.FIELD_ETALON_ID, keys.getEtalonKey().getId()))
                .add(FieldsGroup.and(
                        FormField.range(RecordHeaderField.FIELD_FROM, null, point),
                        FormField.range(RecordHeaderField.FIELD_TO, point, null)));

        SearchRequestContext ctx = SearchRequestContext.builder(EntityIndexType.RECORD, keys.getEntityName())
                .count(1)
                .query(SearchQuery.formQuery(and))
                .source(false)
                .nestedSearch(
                        NestedSearchRequestContext.objects(
                                SearchRequestContext.builder(EntityIndexType.RECORD, keys.getEntityName())
                                    .nestedPath(RecordsDataQualityHeaderField.FIELD_QUALITY_ERRORS.getPath())
                                    .fetchAll(true)
                                    .count(100)
                                    .returnFields(
                                            RecordsDataQualityHeaderField.FIELD_SET_NAME.getPath(),
                                            RecordsDataQualityHeaderField.FIELD_RULE_NAME.getPath(),
                                            RecordsDataQualityHeaderField.FIELD_ERROR.getPath(),
                                            RecordsDataQualityHeaderField.FIELD_SPOT.getPath())
                                    .build())
                                .nestedQueryName(RecordsDataQualityHeaderField.FIELD_QUALITY_ERRORS.getPath())
                                .build())
                .routings(Collections.singletonList(keys.getEtalonKey().getId()))
                .build();

        return searchService.search(ctx);
    }

    private DataQualityResult extractSearch(SearchResultDTO search) {

        DataQualityResult result = new DataQualityResult();
        List<SearchResultHitDTO> innerHits = search.getHits()
                .get(0)
                .getInnerHits()
                .get(RecordsDataQualityHeaderField.FIELD_QUALITY_ERRORS.getPath());

        for (SearchResultHitDTO innerHit : innerHits) {

            String setName = innerHit.getFieldFirstValue(RecordsDataQualityHeaderField.FIELD_SET_NAME.getPath());
            String ruleName = innerHit.getFieldFirstValue(RecordsDataQualityHeaderField.FIELD_RULE_NAME.getPath());

            // We consider DQ errors not relevant anymore,
            // if the set or rule were removed in the mean time.
            MappingSetElement set = metaModelService.instance(DataQualityDescriptors.DQ)
                    .getSet(setName);

            QualityRuleElement rule = metaModelService.instance(DataQualityDescriptors.DQ)
                .getRule(ruleName);

            if (Objects.isNull(set) || Objects.isNull(rule)) {
                continue;
            }

            List<DataQualitySpot> spots =
                innerHit.getFieldValues(RecordsDataQualityHeaderField.FIELD_SPOT.getPath()).stream()
                    .filter(Objects::nonNull)
                    .map(p -> new DataQualitySpot(p.toString(), null))
                    .collect(Collectors.toList());

            List<DataQualityError> errors =
                innerHit.getFieldValues(RecordsDataQualityHeaderField.FIELD_ERROR.getPath()).stream()
                    .map(String.class::cast)
                    .map(DataQualitySerializer::errorFromString)
                    .collect(Collectors.toList());

            RuleExecutionResult ruleResult = new RuleExecutionResult(set, rule);
            ruleResult.addSpots(spots);
            ruleResult.addErrors(errors);
            ruleResult.setEnriched(false);
            ruleResult.setValid(false);

            result.add(ruleResult);
        }

        return result;
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public boolean supports(Start<?, ?> start) {
        return GetRequestContext.class.isAssignableFrom(start.getInputTypeClass());
    }
}
